package com.ibm.spatiotemp.index;

import java.util.Collection;
import java.util.LinkedList;

import org.apache.log4j.Logger;

import com.ibm.spatiotemp.datamodel.datatype.Constraints;
import com.ibm.spatiotemp.datamodel.datatype.TemporalLineString;
import com.ibm.spatiotemp.datamodel.datatype.request.BuildIdxRequest;
import com.ibm.spatiotemp.datamodel.datatype.request.Request;
import com.ibm.spatiotemp.datamodel.datatype.request.RequestPartitioner;
import com.ibm.spatiotemp.datamodel.datatype.request.TSBuildIdxRequest;
import com.ibm.spatiotemp.datamodel.datatype.response.Response;
import com.ibm.spatiotemp.job.IndexJobRunner;
import com.ibm.spatiotemp.job.JobID;
import com.ibm.spatiotemp.job.processor.TSIndexBuildingProcessor;
import com.ibm.spatiotemp.server.config.LocalConfig;
import com.ibm.spatiotemp.source.GPSDataSource;

public class IndexBuildingHandler {

	public static Logger logger = Logger.getLogger(IndexBuildingHandler.class);
	public static String runningMode = Constraints.IDX_LOCALMODE;
	protected JobID generateJObID(){
		return new JobID("testjob");
	}
	
	/*in restful server part, we will implement GPSDataSource as a new class*/
	public Response BuildIndex(BuildIdxRequest request, GPSDataSource dsrc){
		/*todo: add logic to split request into request[], via dsrc's indextimescale
		 * need to implement BuildIdxRequest.clone(), and change its time setting,
		 * start ts, end ts
		 * */
		
		/*step1: find all timeseries id and build request for each ts*/
//		GPSDataSource dsrc = GPSDataSouceFactory.getGPSDataSource(request.getDatasrcid());
		Collection<Request> partitonedreqs = 
				RequestPartitioner.partitionRequestByTimeScale(request, dsrc.getIndexScale());
		
		if(runningMode.equalsIgnoreCase(Constraints.IDX_LOCALMODE)){
		for (Request unitreq : partitonedreqs) {
			BuildIndexOnUnitTimeRange((BuildIdxRequest) unitreq, dsrc);
		}
		}
		else if(runningMode.equalsIgnoreCase(Constraints.IDX_MRMODE))
		{
			logger.error("not support building index via map reduce");
		}
		else if(runningMode.equalsIgnoreCase(Constraints.IDX_SPARKMODE))
		{
			logger.error("not support building index via spark");
		}
		
		return null;
	}

	private void BuildIndexOnUnitTimeRange(BuildIdxRequest request,
			GPSDataSource dsrc) {
		LinkedList<TSBuildIdxRequest> joblist = new LinkedList<>();
		for (String tsid : dsrc.listTimeseriesID()) {
			TSBuildIdxRequest creq = new TSBuildIdxRequest(request,tsid);
			joblist.add(creq);
		} 
		/*step2: parallel processing each request to split gps data, write to hdfs, spark 1*/
		/*step3: parallel processing each splitted linestring set to build Rtreeidx, spark 2*/
		
		Collection<TemporalLineString> linestringset = null;
		JobID jobid =  generateJObID();
		IndexJobRunner jobrunner = new IndexJobRunner();
		try {
			linestringset = jobrunner.runJobs(jobid, dsrc, joblist, LocalConfig.IDX_JOBMODE);
			
			/*linestringset is meta data to be indexed in first index layer*/
			TSIndexBuildingProcessor idxprocesor = new TSIndexBuildingProcessor();
			idxprocesor.writeLinestringset2HDFS(linestringset, jobid.getJobTempPath());
			idxprocesor.buildRtree(jobid.getJobTempPath(), request.getFirstlayeridxpath());
		} catch (Exception e) {
			logger.error(e);
		}
	}
	
}
